package net.java.aioechotest;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ExecutorService;

/**
 * @author Leonid Shlyapnikov
 */
public class AioEchoPoller {

    private static final Logger log = LoggerFactory.getLogger(AioEchoPoller.class);

    private final InetSocketAddress address;

    private final ExecutorService executor;

    private ServerSocketChannel ssc;

    private volatile boolean run = false;

    private Selector selector;

    public AioEchoPoller(int port, ExecutorService executor) {
        this(null, port, executor);
    }

    public AioEchoPoller(String hostname, int port, ExecutorService executor) {
        this.address = (null != hostname ? new InetSocketAddress(hostname, port) :
                new InetSocketAddress(port));
        this.executor = executor;
    }

    public void start() throws IOException {
        init();
        runMainLoop();
    }

    private void init() throws IOException {
        ssc = ServerSocketChannel.open();
        ssc.configureBlocking(false);
        ServerSocket ss = ssc.socket();
        ss.bind(address);

        selector = Selector.open();
        ssc.register(selector, SelectionKey.OP_ACCEPT);
    }

    private void runMainLoop() {
        run = true;

        log.info("Started " + this);

        while (run) {
            try {
                run = runOnePollTurn();
            } catch (Exception e) {
                log.error("Canot finish poll turn", e);
            } catch (Throwable e) {
                final String errMsg = "Canot continue running poll turn";
                log.error(errMsg, e);
                throw new IllegalStateException(errMsg, e);
            }
        }

        log.info("Stopping " + this + "...");
    }

    private boolean runOnePollTurn() throws IOException {
        int num = selector.select();

        if (0 == num) {
            return true;
        }

        final Set<SelectionKey> selectedKeys = selector.selectedKeys();

        for (Iterator<SelectionKey> it = selectedKeys.iterator(); it.hasNext();) {
            final SelectionKey key = it.next();

            if (key.isAcceptable()) {
                // Accept the new connection
                ServerSocketChannel ssc = (ServerSocketChannel) key.channel();
                SocketChannel sc = ssc.accept();
//                executor.execute(new OpAcceptTask(sc, selector, key));
                new OpAcceptTask(sc, selector).run();
                it.remove();
            } else if (key.isReadable()) {
                // Read the data
                SocketChannel sc = (SocketChannel) key.channel();
                executor.execute(new OpReadEchoTask(sc));
//                new OpReadEchoTask(sc).run();
                it.remove();
            }
        }

        return true;
    }
}
